package cn.bearspark.framework.mq.starter.config;

import cn.bearspark.framework.mq.starter.properties.RocketMQConsumerProperties;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

/**
 * RocketMQ 的消费者的抽象父类
 *
 * @author f
 */
public abstract class RocketMQConsumer {

    private final RocketMQConsumerProperties rocketMQConsumerProperties;

    public RocketMQConsumer(RocketMQConsumerProperties rocketMQConsumerProperties) {
        this.rocketMQConsumerProperties = rocketMQConsumerProperties;
    }

    /**
     * 初始化消费者
     *
     * @param topicName          消费的主题名称
     * @param subClassSimpleName 消费者类 (子类) 的简单名称
     * @param listener           消息监听器
     * @throws Exception 在启动消费者时发生的异常
     */
    protected void init(String topicName, String subClassSimpleName,
                        MessageListenerConcurrently listener) throws Exception {
        DefaultMQPushConsumer mqPushConsumer = new DefaultMQPushConsumer();
        // 老版本中会开启，新版本的 MQ 不需要使用到 vip 通道
        mqPushConsumer.setVipChannelEnabled(false);
        // 设置 nameSrv 的地址
        mqPushConsumer.setNamesrvAddr(rocketMQConsumerProperties.getNameSrv());
        // 设置 消费者组的组名
        mqPushConsumer.setConsumerGroup(rocketMQConsumerProperties.getGroupNamePrefix() + "_" + subClassSimpleName);
        // 设置 单次拉取消息进行消费的最大数量
        mqPushConsumer.setConsumeMessageBatchMaxSize(10);
        // 设置 消费者的消费策略，从头开始消费
        mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置 消费者订阅的主题
        mqPushConsumer.subscribe(topicName, "");
        // 设置 消费者的消息监听器
        mqPushConsumer.setMessageListener(listener);
        mqPushConsumer.start();
    }
}
